package IncreatFunction;

import bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author Spring_Hu
 * @date 2021/10/13 20:49
 */
public class IncrFunction_AggregateFunction {
    public static void main(String[] args) throws Exception {
        //增量函数 aggregate
        //1.create 方法 初始化只会在第一条数据来的时候才会调用 只调用一次 （时间驱动）
        //2.add 来一条处理一条
        //3.获取结果只有一次 结果传递给下游 （此处为print）

        //reduce 与aggregated的区别 1.reduce第一条不处理 2.aggregate比reduce更灵活 reduce要求输入类型 中间状态
        // 输出类型都要一致 aggregate可以三者不一致

        //全量函数 就是process函数 其是将同一个窗口的数据都收集存起来 然后在窗口关闭后将结果传递给下游

        //为何要用到全量函数？  TopN 的需求实现  适合使用全量函数




        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("hadoop102",8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
                    }
                })
                .keyBy(value -> value.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
                    @Override
                    public Integer createAccumulator() {
                        System.out.println("creat-----");
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor value, Integer accumulator) {
                        System.out.println("add------");
                        return value.getVc()+accumulator;
                    }

                    @Override
                    public String getResult(Integer accumulator) {
                        return "累加结果"+accumulator.toString();
                    }

                    //只有会话窗口才会调用 回话是基于时间的 底层是用固定的时间窗口 合并成间隔时间条件的窗口
                    // 此方法是在合并窗口的时候调用
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        System.out.println("merge---------------");
                        return null;
                    }
                })
                .print("print");

        env.execute();
    }
}
